package com.github.slprj.slake.net;

import com.github.slprj.slake.common.SlTime;
import com.github.slprj.slake.log.SlLog;
import com.github.slprj.slake.log.SlLogs;

import java.util.ArrayList;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * @author YangXiaoLong
 * 2022/9/24 16:49
 */
public class NioClient extends NioNet {
    public static final SlLog log = SlLogs.getLogger(NioClient.class);
    private final NioClientConfig config;
    private final NioWorker worker;
    //用于发送消息的负载均衡
    private int invokeIndex;

    public NioClient(NioClientConfig config) {
        super(config);
        this.config = config;
        if (config.getHostPorts() == null || config.getHostPorts().size() == 0) {
            throw new IllegalArgumentException("no servers");
        }
        worker = new NioWorker(nioStatus, config.getName() + "IoWorker", config);
    }

    @Override
    protected void doStart() throws Exception {
        worker.start();
        //1.建立连接
        SlTime t = new SlTime(config.getConnectTimeoutMillis(), TimeUnit.MILLISECONDS);
        ArrayList<CompletableFuture<SlChannel>> futures = new ArrayList<>();
        for (HostPort hp : config.getHostPorts()) {
            CompletableFuture<SlChannel> f = worker.connect(hp);
            futures.add(f);
        }

        //2.统计成功与失败的数量
        for (CompletableFuture<SlChannel> f : futures) {
            long restMillis = t.rest(TimeUnit.MILLISECONDS);
            //等待超时时间内连接完成
            if (restMillis > 0 && !Thread.currentThread().isInterrupted()) {
                try {
                    f.get(restMillis, TimeUnit.MILLISECONDS);
                } catch (Exception e) {
                    // ignore
                    log.info("等待连接异常");
                }
            }
        }
        int successCount = 0;
        int timeoutCount = 0;
        int failCount = 0;

        for (CompletableFuture<SlChannel> f : futures) {
            if (f.isDone()) {
                if (f.isCompletedExceptionally() || f.isCancelled()) {
                    failCount++;
                } else {
                    successCount++;
                }
            } else {
                timeoutCount++;
            }
        }

        if (successCount == 0) {
            throw new NetException("init NioClient fail:timeout=" + config.getConnectTimeoutMillis()
                    + "ms, timeoutConnectionCount=" + timeoutCount + ", failConnectionCount=" + failCount);
        }
        initBizExecutor();
    }

    public CompletableFuture<ReadFrame> sendRequest(WriteFrame request, Decoder decoder, SlTime timeout) {
        return sendRequest(worker, null, request, decoder, timeout);
    }

    public CompletableFuture<ReadFrame> sendRequest(Peer peer, WriteFrame request, Decoder decoder, SlTime timeout) {
        return sendRequest(worker, peer, request, decoder, timeout);
    }

}
